package com.tl.job014.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * kafka生产者工具类
 * 
 * @author yingfing
 * @date 2021年1月20日 @time 下午10:51:02
 */
public class KafkaProducerUtil4Java {

	public KafkaProducer<String, String> producer = null;

	public KafkaProducerUtil4Java(String bootStrapServer) {
		// 通过Properties工具类封装kafka自提供KafkaProducer的参数
		Properties prop = new Properties();

		prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServer);
		prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				StringSerializer.class.getName());
		prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				StringSerializer.class.getName());

		producer = new KafkaProducer<String, String>(prop);
	}

	public static void main(String[] args) {

		// 1、传入bootstrap-server参数
		String bootStrapServer = "cluster1.hadoop:6667,clusterkk::6667";
		String topic = "zyfKafkaTest";

		// 2、构建KafkaProducerUtil4Java对象
		KafkaProducerUtil4Java producerUtil = new KafkaProducerUtil4Java(
				bootStrapServer);

		// 3、生产消息发送到指定的topic
		producerUtil.producer.send(new ProducerRecord<String, String>(topic,
				"HelloWorld,job014大数据14期！"));

		// 4、关闭链接
		producerUtil.producer.close();
	}
}
